/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.dtub.mongo.update.worker;

import cn.ac.iie.di.ban.dtub.mongo.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.idgen.IDSHAGeneratorImpl;
import cn.ac.iie.di.ban.dtub.mongo.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author zwlin
 */
public class ReceiveHandler extends REAbstractReceiveMessageHandler<REMessage> {

    static Logger logger = null;

    static {
        PropertyConfigurator.configure("./log4j.properties");
        logger = Logger.getLogger(ReceiveHandler.class.getName());
    }

    static private final HashMap<String, REFieldType> types;

    static {

        types = new HashMap<>();

        types.put("_id", REFieldType.String);
        types.put("_key", REFieldType.String);
        types.put("_spec", REFieldType.String);
        types.put("_ch", REFieldType.Long);
        types.put("ut", REFieldType.Long);
        types.put("nrd", REFieldType.Long);
    }
    static private final String[] lookupFields = {"_id", "_key", "_spec", "_ch", "url"};

    private boolean stopped;
    private final AtomicInteger aliveCount;

    private final MongoHelper[] mongos;
    private final Random rand;
    private final SenderPool[] pools;
    ArrayBlockingQueue<HashMap<String, Object>> queue;
    static private IDSHAGeneratorImpl idgenerator = new IDSHAGeneratorImpl();

    static private String getId(String wx_uid) {
        try {
            return idgenerator.getId("40", "U-WX03-A", "WeChat", "WeChat", wx_uid);
        } catch (Exception ex) {
            return null;
        }
    }

    public ReceiveHandler(MongoHelper[] mongos, SenderPool[] pools, ArrayBlockingQueue<HashMap<String, Object>> queue) {
        this.mongos = mongos;
        this.rand = new Random(System.currentTimeMillis());
        this.pools = pools;
        stopped = false;
        aliveCount = new AtomicInteger(0);
        this.queue = queue;
    }

    @Override
    public boolean handle(REMessage message) {
        //env
        if (stopped) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException ex) {
            }
            return false;
        }
        logger.info("receive record size: " + message.data.size());
        try {
            //get idx;
            int wxIdx = -1;
            int nrdIdx = -1;
            for (int i = 0; i < message.colNames.length; ++i) {
                if ("wx_sfrom".equals(message.colNames[i])) {
                    wxIdx = i;
                } else if ("nrd".equals(message.colNames[i])) {
                    nrdIdx = i;
                } else if (wxIdx >= 0 && nrdIdx >= 0) {
                    break;
                }
            }
            if (wxIdx < 0) {
                logger.error("wx_sfrom column not exist!");
                return true;
            }
            if (nrdIdx < 0) {
                logger.error("nrd column not exist!");
                return true;
            }

            for (Object[] rec : message.data) {
                if (rec[nrdIdx] == null) {
                    continue;
                }
                String wx_sfrom = (String) rec[wxIdx];
                String _id = getId(wx_sfrom);
//                logger.info("_id : " + _id);
                Long nrd = (Long) rec[nrdIdx];

                HashMap<String, Object> urlMap = new HashMap<>();
                urlMap.put("_id", _id);
                urlMap.put("nrd", nrd);

                while (true) {
                    try {
                        queue.put(urlMap);
                        break;
                    } catch (InterruptedException ex) {
                        logger.warn("put into queue error", ex);
                    }
                }

            }

        } catch (Exception e) {
            logger.error("some thing error occur when do delayed init!", e);
        } finally {
            aliveCount.getAndDecrement();
        }
        return true;
    }

    public void stopGracefully() {
        stopped = true;
        while (aliveCount.get() > 0) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException ex) {
                logger.warn("sleep interrupted.", ex);
            }
        }
    }
}
